package ETL

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	_ "github.com/denisenkom/go-mssqldb"
	_ "github.com/go-sql-driver/mysql"
	_ "github.com/godror/godror"
	"github.com/pkg/errors"
	"db2s/ref"
	"sort"
	"strconv"
	"strings"
	"time"
)

const (
	Map                   = "map"
	String                = "string"
	StringPoint           = "*string"
	SMap                  = "sMap"
	StringGarth           = "[]string"
	StringPointGarth      = "[]*string"
	GarthStringPointGarth = "[][]*string"
	FlowAny               = "chanS"
	FlowString            = "chanString"
	DataRowsSplit         = "/*go actions rowData*/"
	DataColumnsSplit      = "/*go actions columnData*/"
)

type InstanceActiveS struct {
	SqlRows         *sql.Rows
	Schema          string
	Table           string
	TableColumnType []map[string]string
	DB              *sql.DB
	DBType          string
	Tx              *sql.Tx
	StopTime        time.Duration
	SqlLogBin       bool
	SqlMode         bool
	SessionSet      []string
	LobSwitch       bool
	FixRows         int64
	InsertPreFix    string
	Expect          string
	GisValFunc      func(any) any
}
type FnReturn struct {
	ExecTime int64
	Err      error
	Result   any
}

// columnTypeActive 处理单列数据类型
func (dbActive InstanceActiveS) columnTypeActive(columnName string) (string, int, bool) {
	var (
		indexColumnType   string //索引列的数据类型
		indexColumnIsNull bool   //索引列数据类型是否允许为null
		office            int    //浮点类型的偏移量
	)
	IntType := []string{"TINYINT", "SMALLINT", "MEDIUMINT", "INT", "BIGINT"}
	floatType := []string{"FLOAT", "DOUBLE", "DECIMAL"}
	for _, i := range dbActive.TableColumnType {
		if i["columnName"] == columnName {
			ct := strings.Split(strings.ToUpper(i["dataType"]), "(")[0]
			if strings.Contains(strings.Join(IntType, ","), ct) {
				indexColumnType = "int"
			} else if strings.Contains(strings.Join(floatType, ","), ct) {
				office, _ = strconv.Atoi(strings.TrimSpace(strings.ReplaceAll(strings.Split(i["dataType"], ",")[1], ")", "")))
				indexColumnType = "float"
			} else {
				indexColumnType = "char"
			}
			if i["isNull"] == "YES" { //判断当前索引列是否允许为null
				indexColumnIsNull = true
			}
		}
	}
	return indexColumnType, office, indexColumnIsNull
}

func (dbActive InstanceActiveS) dataAscSort(data map[string]interface{}, columnName string) []map[string]string {
	var (
		indexColumnUniqueS []map[string]string
		z                  = make(map[string]int)         //源目标端索引列数据的集合（无序的）
		zInt               []int                          //int类型的索引列集合，用于正序排序
		zFloat             []float64                      //float类型的索引列集合，用于正序排序
		zChar              []string                       //char类型的索引列集合，用于正序排序
		zNull              = make(map[string]interface{}) //针对null的值的一个处理
		office             int                            //浮点类型的偏移量
	)
	//原目标端索引列数据去重，并按照索引列数据类型进行分类合并索引列
	indexColumnType, office, indexColumnIsNull := dbActive.columnTypeActive(columnName)
	for k, v := range data {
		switch indexColumnType {
		case "int":
			if indexColumnIsNull {
				if k != "<nil>" {
					zc, _ := strconv.Atoi(k)
					zInt = append(zInt, zc)
				} else {
					zNull["<nil>"] = v
				}
			} else {
				zc, _ := strconv.Atoi(k)
				zInt = append(zInt, zc)
			}
		case "float":
			//处理null值
			if indexColumnIsNull {
				if k != "<nil>" {
					zc, _ := strconv.ParseFloat(k, office)
					zFloat = append(zFloat, zc)
				} else {
					zNull["<nil>"] = v
				}
			} else {
				zc, _ := strconv.ParseFloat(k, office)
				zFloat = append(zFloat, zc)
			}
		case "char":
			if indexColumnIsNull {
				if k != "<nil>" {
					zChar = append(zChar, k)
				} else {
					zNull["<nil>"] = v
				}
			} else {
				zChar = append(zChar, k)
			}
		}
	}
	//针对索引类数据进行正序排序
	switch indexColumnType {
	case "int":
		sort.Ints(zInt)
		if _, ok := zNull["<nil>"]; ok {
			indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{"columnName": fmt.Sprintf("%v", "<nil>"), "count": fmt.Sprintf("%v", z[fmt.Sprintf("%v", "<nil>")])})
		} else {
			for _, i := range zInt {
				indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{fmt.Sprintf("%v", i): fmt.Sprintf("%v", data[fmt.Sprintf("%v", i)])})
			}
		}
		zInt, z = nil, nil

	case "float":
		sort.Float64s(zFloat)
		if _, ok := zNull["<nil>"]; ok {
			indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{"columnName": fmt.Sprintf("%v", "<nil>"), "count": fmt.Sprintf("%v", z[fmt.Sprintf("%v", "<nil>")])})
		} else {
			for _, i := range zFloat {
				ii := strconv.FormatFloat(i, 'f', 2, 64)
				indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{ii: fmt.Sprintf("%v", data[ii])})
			}
		}
		zFloat, z = nil, nil
	case "char":
		sort.Strings(zChar)
		if _, ok := zNull["<nil>"]; ok {
			indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{"columnName": fmt.Sprintf("%v", "<nil>"), "count": fmt.Sprintf("%v", z[fmt.Sprintf("%v", "<nil>")])})
		} else {
			for _, i := range zChar {
				indexColumnUniqueS = append(indexColumnUniqueS, map[string]string{i: fmt.Sprintf("%v", data[i])})
			}
		}
		zChar, z = nil, nil
	}
	return indexColumnUniqueS
}

// rowsDataNullActive 处理行数据的null值
func (dbActive InstanceActiveS) rowsDataNullActive(i map[string]interface{}) map[string]interface{} {
	var (
		zNull = make(map[string]interface{}) //源目标端索引列数据的集合（无序的） //针对null的值的一个处理
	)
	if fmt.Sprintf("%v", i["columnName"]) == "<nil>" {
		if _, ok := i["count"]; ok {
			c, _ := strconv.ParseUint(fmt.Sprintf("%v", i["count"]), 10, 64)
			zNull["<nil>"] = c
		}
	} else {
		zNull[fmt.Sprintf("%v", i["columnName"])] = i["count"]
	}
	return zNull
}

// sql语句循环执行，循环执行3次
func (dbActive InstanceActiveS) forExecQuerySql1(ctx context.Context, trx any, strSql string) (rows *sql.Rows, err error) {
	// 执行包含可能导致错误的操作
	var event = "[forExecQuerySql1]"
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	for i := 1; i < 4; i++ {
		switch trx.(type) {
		case *sql.Tx:
			tr := trx.(*sql.Tx)
			rows, err = tr.Query(strSql)
		case *sql.DB:
			tr := trx.(*sql.DB)
			if ctx != nil {
				rows, err = tr.QueryContext(ctx, strSql)
			} else {
				rows, err = tr.Query(strSql)
			}
		}
		if err != nil {
			if i == 3 {
				err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("exec sql fail. error info is {%v}", err)))
				return
			}
			time.Sleep(1 * time.Second)
		} else {
			break
		}
	}
	return
}

// getColumnName 获取列名
func (dbActive InstanceActiveS) getColumnName(SqlRows *sql.Rows) (columns []string, err error) {
	var event = "[getColumnName]"
	if columns, err = SqlRows.Columns(); err != nil {
		err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("DB Get the column fail. %v", err)))
		return
	}
	return
}

// entryActive 行数据中空字符串处理
func (dbActive InstanceActiveS) entryActive(dbType string, v any) (res any) {
	if dbActive.GisValFunc != nil {
		res = dbActive.GisValFunc(v)
	}
	return
}

// oneRowsActive 单行数据多个列进行处理
func (dbActive InstanceActiveS) oneRowsActive(dbType string, columns []string, values []interface{}, outPutType string) (res any) {
	var (
		stringJoin bytes.Buffer
		mapJoin    = make(map[string]interface{})
		sliceJoin  []*string
	)
	columnLength := len(columns)
	for i, k := range columns {
		var v interface{}
		val := values[i]
		b, ok := val.([]byte)
		if ok {
			v = string(b)
		} else {
			v = val
		}
		v = dbActive.entryActive(dbType, v)
		switch outPutType {
		case String, StringPoint:
			stringJoin.WriteString(fmt.Sprintf("%v", v))
			if i < columnLength-1 {
				stringJoin.WriteString(DataColumnsSplit)
			}
		case StringPointGarth:
			l := fmt.Sprintf("%v", v)
			sliceJoin = append(sliceJoin, &l)
		case Map, SMap:
			mapJoin[k] = v
		}
	}
	switch outPutType {
	case String:
		if l := stringJoin.String(); len(l) > 0 {
			res = l
		}
		return
	case StringPoint:
		if l := stringJoin.String(); len(l) > 0 {
			res = &l
		}
		return
	case StringPointGarth:
		if len(sliceJoin) > 0 {
			res = sliceJoin
		}
		return
	case GarthStringPointGarth:
		return sliceJoin
	case Map, SMap:
		if len(mapJoin) > 0 {
			res = mapJoin
		}
		return
	}
	return
}

// manyRowsActive 多行数据进行处理
func (dbActive InstanceActiveS) manyRowsActive(dbType string, SqlRows *sql.Rows, columns []string, outPutType string) (res any, err error) {
	var (
		stringJoin             bytes.Buffer
		mapJoin                = make(map[string]interface{})
		sliceGather            []string
		sliceStringPointGather []*string
		slicePointGather       [][]*string
		ff                     []map[string]interface{}
		event                  = "[manyRowsActive]"
	)
	// 定义一个切片，长度是字段的个数，切片里面的元素类型是sql.RawBytes
	//values := make([]sql.RawBytes,len(columns))
	//定义一个切片，元素类型是interface{}接口
	//scanArgs := make([]interface{},len(values))
	valueStr := make([]interface{}, len(columns))
	values := make([]interface{}, len(columns))
	for SqlRows.Next() {
		for i := 0; i < len(columns); i++ {
			valueStr[i] = &values[i]
		}
		if err = SqlRows.Scan(valueStr...); err != nil {
			err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("DB Get the rows data fail. %v", err)))
			return
		}
		switch outPutType {
		case String, StringPoint:
			if q := dbActive.oneRowsActive(dbType, columns, values, String); q != nil {
				stringJoin.WriteString(q.(string))
				stringJoin.WriteString(DataRowsSplit)
			}
		case Map:
			if ll := dbActive.oneRowsActive(dbType, columns, values, outPutType); ll != nil {
				mapJoin = ll.(map[string]interface{})
			}
		case StringGarth:
			if ll := dbActive.oneRowsActive(dbType, columns, values, String); ll != nil {
				sliceGather = append(sliceGather, ll.(string))
			}
		case StringPointGarth:
			if ll := dbActive.oneRowsActive(dbType, columns, values, StringPoint); ll != nil {
				sliceStringPointGather = append(sliceStringPointGather, ll.(*string))
			}
		case GarthStringPointGarth:
			if ll := dbActive.oneRowsActive(dbType, columns, values, StringPointGarth); ll != nil {
				slicePointGather = append(slicePointGather, ll.([]*string))
			}
		case SMap:
			if ll := dbActive.oneRowsActive(dbType, columns, values, Map); ll != nil {
				ff = append(ff, ll.(map[string]interface{}))
			}
		default:
			err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("type not match. curry type is %v", outPutType)))
			return
		}
	}
	switch outPutType {
	case String:
		if q := stringJoin.String(); len(q) > 0 {
			res = q
		}
		return
	case StringPoint:
		if q := stringJoin.String(); len(q) > 0 {
			res = &q
		}
		return
	case StringGarth:
		if len(sliceGather) > 0 {
			res = sliceGather
		}
		return
	case StringPointGarth:
		if len(sliceStringPointGather) > 0 {
			res = sliceStringPointGather
		}
		return
	case Map:
		if len(mapJoin) > 0 {
			res = mapJoin
		}
		return
	case SMap:
		if len(ff) > 0 {
			res = ff
		}
		return
	case GarthStringPointGarth:
		if len(slicePointGather) > 0 {
			res = slicePointGather
		}
		return
	default:
		err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("type not match. curry type is %v", outPutType)))
		return
	}
}

// rowsColumnActiveString  rows处理，返回字符串
func (dbActive InstanceActiveS) rowsColumnActiveString(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[rowsColumnActiveString]"
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res, err = dbActive.manyRowsActive(dbType, SqlRows, columns, String); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res == nil {
		return
	}
	if strings.HasSuffix(res.(string), DataRowsSplit) {
		if n := strings.LastIndex(res.(string), DataRowsSplit); n != -1 {
			return res.(string)[:n], err
		}
	}
	return
}

// rowsColumnActiveStringPoint  rows处理，返回字符串
func (dbActive InstanceActiveS) rowsColumnActiveStringPoint(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[rowsColumnActiveStringPoint]"
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res, err = dbActive.manyRowsActive(dbType, SqlRows, columns, StringPoint); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res == nil {
		return
	}
	rowD := *res.(*string)
	if strings.HasSuffix(rowD, DataRowsSplit) {
		if n := strings.LastIndex(rowD, DataRowsSplit); n != -1 {
			rowD = rowD[:n]
		}
	}
	res = &rowD
	return
}

// rowsColumnActiveSliceString  rows处理，返回字符串
func (dbActive InstanceActiveS) rowsColumnActiveSliceString(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[rowsColumnActiveSliceString]"
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return dbActive.manyRowsActive(dbType, SqlRows, columns, StringGarth)
}

// rowsColumnActiveSliceStringPoint  rows处理，返回字符串
func (dbActive InstanceActiveS) rowsColumnActiveSliceStringPoint(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[rowsColumnActiveSliceStringPoint]"
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res, err = dbActive.manyRowsActive(dbType, SqlRows, columns, GarthStringPointGarth); err != nil || res == nil {
		return
	}
	return
}

// rowsActiveMap 数据返回map
func (dbActive InstanceActiveS) rowsActiveMap(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[rowsActiveMap]"
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return dbActive.manyRowsActive(dbType, SqlRows, columns, Map)
}

// manyRowsActiveSMap  处理多行数据返回map
func (dbActive InstanceActiveS) manyRowsActiveSMap(dbType string, SqlRows *sql.Rows) (res any, err error) {
	var columns []string
	var event = "[manyRowsActiveSMap]"
	defer func() {
		if err = rowsClose(SqlRows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	}()
	if columns, err = dbActive.getColumnName(SqlRows); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return dbActive.manyRowsActive(dbType, SqlRows, columns, SMap)
}

func (dbActive InstanceActiveS) chanRowsActiveS1(dbType string, SqlRows *sql.Rows) chan any {
	var chanEntry = make(chan any, dbActive.FixRows)
	go func() {
		var (
			columns    []string
			err        error
			maxRetries = 1           // 最大重试次数
			retryDelay = time.Second // 重试延迟
			si         int
		)
		defer func() {
			close(chanEntry)
		}()
		for attempts := 0; attempts < maxRetries; attempts++ {
			if columns, err = dbActive.getColumnName(SqlRows); err != nil {
				time.Sleep(retryDelay) // 等待后重试
				continue
			}
			// 检查是否有数据
			if !SqlRows.Next() {
				//err = errors.New(fmt.Sprintf("SqlRows.Next is nil"))
				return
			}
			valueStr := make([]interface{}, len(columns))
			values := make([]interface{}, len(columns))
			for i := 0; i < len(columns); i++ {
				valueStr[i] = &values[i]
			}
			if err = SqlRows.Scan(valueStr...); err != nil {
				time.Sleep(retryDelay) // 等待后重试
				continue
			}
			s1 := dbActive.oneRowsActive(dbType, columns, values, StringPointGarth)
			chanEntry <- s1
			// 处理余下行数据
			for SqlRows.Next() {
				for i := 0; i < len(columns); i++ {
					valueStr[i] = &values[i]
				}
				if err = SqlRows.Scan(valueStr...); err != nil {
					time.Sleep(retryDelay) // 等待后重试
					continue
				}
				s2 := dbActive.oneRowsActive(dbType, columns, values, StringPointGarth)
				si++
				chanEntry <- s2
			}
			if err = SqlRows.Close(); err != nil {
				err = errors.New(fmt.Sprintf("DB rows close fail. error is %v", err))
				return
			}
			return // 成功执行完毕
		}
		// 如果达到最大重试次数，处理错误
		//err = errors.New(fmt.Sprintf("Database query failed after multiple attempts.err is %v", err))
		// 你可以选择在这里发送错误信息到通道或进行其他处理
	}()
	return chanEntry
}

// chanRowsActiveS  处理多行数据返回map
func (dbActive InstanceActiveS) chanRowsActiveS(dbType string, SqlRows *sql.Rows) chan any {
	var chanEntry = make(chan any, dbActive.FixRows)
	go func() {
		var (
			columns []string
			err     error
		)
		if columns, err = dbActive.getColumnName(SqlRows); err != nil {
			return
		}
		// 定义一个切片，长度是字段的个数，切片里面的元素类型是sql.RawBytes
		//values := make([]sql.RawBytes,len(columns))
		//定义一个切片，元素类型是interface{}接口
		//scanArgs := make([]interface{},len(values))
		valueStr := make([]interface{}, len(columns))
		values := make([]interface{}, len(columns))
		for SqlRows.Next() {
			for i := 0; i < len(columns); i++ {
				valueStr[i] = &values[i]
			}
			if err = SqlRows.Scan(valueStr...); err != nil {
				return
			}
			s1 := dbActive.oneRowsActive(dbType, columns, values, StringPointGarth)
			chanEntry <- s1
		}
		close(chanEntry)
		if err = SqlRows.Close(); err != nil {
			err = errors.New(fmt.Sprintf("DB rows close fail. error is %v", err))
			return
		}
	}()
	return chanEntry
}

// DataChanActive /*
func (dbActive InstanceActiveS) DataChanActive() chan map[string]interface{} {
	var chanEntry = make(chan map[string]interface{}, 1000)
	go func() {
		columns, err := dbActive.SqlRows.Columns()
		if err != nil {
			err = errors.New(fmt.Sprintf("Get the column fail. error is %v", err))
			return
		}
		// 定义一个切片，长度是字段的个数，切片里面的元素类型是sql.RawBytes
		//values := make([]sql.RawBytes,len(columns))
		//定义一个切片，元素类型是interface{}接口
		//scanArgs := make([]interface{},len(values))
		valueStr := make([]interface{}, len(columns))
		values := make([]interface{}, len(columns))
		for dbActive.SqlRows.Next() {
			for i := 0; i < len(columns); i++ {
				valueStr[i] = &values[i]
			}
			if err := dbActive.SqlRows.Scan(valueStr...); err != nil {
				err = errors.New(fmt.Sprintf("DB Get the rows data fail. error is %v", err))
				return
			}
			entry := make(map[string]interface{})
			for i, col := range columns {
				var v interface{}
				val := values[i]
				b, ok := val.([]byte)
				if ok {
					v = string(b)
				} else {
					v = val
				}
				if v == nil {
					v = "<nil>"
				}
				if v == "" {
					v = "<entry>"
				}
				entry[col] = v
			}
			chanEntry <- dbActive.rowsDataNullActive(entry)
		}
		close(chanEntry)
		if err := dbActive.SqlRows.Close(); err != nil {
			err = errors.New(fmt.Sprintf("DB rows close fail. error is %v", err))
			return
		}
	}()
	// 获取列名
	return chanEntry
}

// SqlQuery sql语句查询 外部调用
func (dbActive InstanceActiveS) SqlQuery(strSql string, rowsType string) (execTime int64, entry any, err error) {
	var (
		rows   *sql.Rows
		ctx    context.Context
		cancel context.CancelFunc
		event  = "[SqlQuery]"
	)
	beginTime := time.Now()
	if dbActive.StopTime > 0 {
		ctx, cancel = context.WithTimeout(context.Background(), dbActive.StopTime)
		defer func() {
			if strings.EqualFold(strings.TrimSpace(fmt.Sprintf("%v", ctx.Err())), fmt.Sprintf("context deadline exceeded")) {
				err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("sql exec timeout")))
				cancel()
			}
		}()
	} else {
		ctx = nil
	}
	if !dbActive.SqlMode {
		if _, err = dbActive.DB.Exec("set session sql_mode=''"); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	}
	if rows, err = dbActive.forExecQuerySql1(ctx, dbActive.DB, strSql); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	defer func() {
		if rowsType != FlowAny {
			if err = rowsClose(rows); err != nil {
				err = ref.ErrAddPrintf(event, err)
				return
			}
		}
	}()
	switch rowsType {
	case Map:
		if entry, err = dbActive.rowsActiveMap(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case String:
		if entry, err = dbActive.rowsColumnActiveString(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case StringPoint:
		if entry, err = dbActive.rowsColumnActiveStringPoint(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case StringGarth:
		if entry, err = dbActive.rowsColumnActiveSliceString(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case GarthStringPointGarth:
		if entry, err = dbActive.rowsColumnActiveSliceStringPoint(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case SMap:
		if entry, err = dbActive.manyRowsActiveSMap(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	case FlowAny:
		if entry = dbActive.chanRowsActiveS1(dbActive.DBType, rows); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	}
	execTime = time.Since(beginTime).Milliseconds()
	return
}

// BeginTrxQuery 开启一个参数
func (dbActive InstanceActiveS) BeginTrxQuery(setVariables []string, strSql string, rowsType string) (execTime int64, entry any, err error) {
	var (
		trx    *sql.Tx
		rows   *sql.Rows
		ctx    context.Context
		cancel context.CancelFunc
	)
	defer func() {
		if r := recover(); r != nil {
			fmt.Println(fmt.Sprintf("BeginTrxQuery Recovered from panic: %v", r))
		}
	}()
	beginTime := time.Now()
	if !dbActive.SqlMode {
		if _, err = dbActive.DB.Exec("set session sql_mode=''"); err != nil {
			return 0, nil, err
		}
	}
	if dbActive.StopTime > 0 {
		ctx, cancel = context.WithTimeout(context.Background(), dbActive.StopTime)
		defer cancel()
		if trx, err = dbActive.DB.BeginTx(ctx, &sql.TxOptions{
			Isolation: sql.LevelReadUncommitted, // 设置事务隔离级别
			ReadOnly:  false,                    // 设置事务为读写模式
		}); err != nil {
			err = errors.New(fmt.Sprintf("context_withTimeout failed to open transaction,context timeout is {%v}. error is {%v}", dbActive.StopTime, err))
			return
		}
	} else {
		if trx, err = beginTrx(dbActive.DB); err != nil {
			err = errors.New(fmt.Sprintf("failed to open transaction. error is {%v}", err))
			return
		}
	}
	if len(setVariables) > 0 {
		for _, v := range setVariables {
			if err = setSessionVariables(ctx, trx, v); err != nil {
				return
			}
		}
	}
	if rows, err = dbActive.forExecQuerySql1(ctx, trx, strSql); err != nil {
		if err1 := trx.Rollback(); err1 != nil {
			err = errors.New(fmt.Sprintf("%v trx rollback fail. error is %v", err, err1))
		}
		return
	}
	switch rowsType {
	case Map:
		if entry, err = dbActive.rowsActiveMap(dbActive.DBType, rows); err != nil {
			return
		}
	case String:
		if entry, err = dbActive.rowsColumnActiveString(dbActive.DBType, rows); err != nil {
			return
		}
	case StringPoint:
		if entry, err = dbActive.rowsColumnActiveStringPoint(dbActive.DBType, rows); err != nil {
			return
		}
	case StringGarth:
		if entry, err = dbActive.rowsColumnActiveSliceString(dbActive.DBType, rows); err != nil {
			return
		}
	case StringPointGarth:
		if entry, err = dbActive.rowsColumnActiveSliceStringPoint(dbActive.DBType, rows); err != nil {
			return
		}
	case SMap:
		if entry, err = dbActive.manyRowsActiveSMap(dbActive.DBType, rows); err != nil {
			return
		}
	case FlowString:

	}
	if err = rowsClose(rows); err != nil {
		return
	}
	if err = trxCommit(trx); err != nil {
		return
	}
	execTime = time.Since(beginTime).Milliseconds()
	return
}
func (dbActive InstanceActiveS) DDL(setVariables any) (execTime int64, err error) {
	//p := RetriesExecSql(dbActive.DDLUpdate, setVariables)
	//return p.ExecTime, p.Err
	return
}
func (dbActive InstanceActiveS) TrxUpdate(setVariables any) (execTime int64, err error) {
	//p := RetriesExecSql(dbActive.BeginTrxUpdate, setVariables)
	//return p.ExecTime, p.Err
	return
}
func (dbActive InstanceActiveS) Update(setVariables any) (execTime int64, err error) {
	//var event = "[Update]"
	//defer func() {
	//	if re := recover(); re != nil {
	//		err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, re))
	//	}
	//}()
	//p := RetriesExecSql(dbActive.normalUpdate, setVariables)
	//return p.ExecTime, p.Err
	return
}
func (dbActive InstanceActiveS) TrxInsert(setVariables any) (execTime int64, err error) {
	//p := RetriesExecSql(dbActive.BeginTrxInsert, setVariables)
	//return p.ExecTime, p.Err
	return
}
